package com.sfx.concurrent.demo.BlockingQueue;

import java.util.Arrays;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * Created with IntelliJ IDEA.
 * Description: 单锁实现阻塞队列
 * User: sfx
 * Date: 2023-05-10
 * Time: 11:00
 */
public class BlockingQueueDoubleLock<E> implements BlockingQueue<E>{
    //头部
    private int head;
    //尾部
    private int tail;
    //队列中实际元素的个数
    private int size;
    //数组
    private E[] queue;

    private int capacity;

    public BlockingQueueDoubleLock(int capacity) {
        this.capacity = capacity;
        queue = (E[]) new Object[capacity];
    }

    //锁对象
    private ReentrantLock lock = new ReentrantLock();
    //当队列满的时候进入的条件变量(tailWaits)
    private Condition tailWaits = lock.newCondition();
    //当队列空的时候进入的条件变量(headWaits)
    private Condition headWaits = lock.newCondition();

    @Override
    public void offer(E e) throws InterruptedException {
        lock.lockInterruptibly();
        try{
            while(isFull()) {
                tailWaits.await();
            }
            this.queue[tail] = e;
            if(++tail == queue.length) {
                tail = 0;
            }
            size ++;
            headWaits.signal();
        }finally {
            lock.unlock();
        }
    }

    @Override
    public boolean offer(E e, long timeout) throws InterruptedException {
        lock.lockInterruptibly();
        long nanos = TimeUnit.MILLISECONDS.toNanos(timeout);
        try{
            while(isFull()) {
                if(nanos <=0) {
                    return  false;
                }
                nanos = tailWaits.awaitNanos(nanos);//返回剩余的等待时间
            }
            queue[tail] = e;
            if(++tail == queue.length) {
                tail = 0;
            }
            size ++;
            headWaits.signal();
            return true;
        }finally {
            lock.unlock();
        }
    }

    @Override
    public E poll() throws InterruptedException {
        lock.lockInterruptibly();
        try{
            while(isEmpty()) {
                headWaits.await();
            }
            E e = queue[head];
            queue[head] = null;
            if(++head == queue.length) {
                head = 0;
            }
            size --;
            tailWaits.signal();
            return e;
        }finally {
            lock.unlock();
        }
    }

    private boolean isEmpty() {
        return this.size == 0;
    }

    private boolean isFull() {
        return this.size == this.capacity;
    }

    @Override
    public String toString() {
        return "MyBlockingQueue{" +
                "queue=" + Arrays.toString(queue);
    }
}